/*
    SUSAN® - Sole of Unix Save ANything

   Copyright (C) 2016-2016 Skyatlas Co. CN

   
*/

/*
 * Connection and Connection Pool
 */

#include "include/susan.h"
#include "connection_pool.h"
#include "lib/util.h"
#include "lib/alist.h"
#include "lib/bsys.h"
#include "lib/bsock.h"

/*
 * Connection
 */
Connection::Connection(const char* name,
                       int protocol_version,
                       SusanSocket* socket,
                       bool authenticated)
{
  tid_ = pthread_self();
  connect_time_ = time(nullptr);
  in_use_ = false;
  authenticated_ = authenticated;
  bstrncpy(name_, name, sizeof(name_));
  protocol_version_ = protocol_version;
  socket_ = socket;
  pthread_mutex_init(&mutex_, nullptr);
}

Connection::~Connection() { pthread_mutex_destroy(&mutex_); }

/*
 * Check if connection is still active.
 */
bool Connection::check(int timeout_data)
{
  int data_available = 0;
  bool ok = true;

  /*
   * Returns: 1 if data available, 0 if timeout, -1 if error
   */
  data_available = socket_->WaitDataIntr(timeout_data);

  /*
   * Use lock to prevent that data is read for job thread.
   */
  lock();
  if (data_available < 0) {
    ok = false;
  } else if ((data_available > 0) && (!in_use_)) {
    if (socket_->recv() <= 0) { ok = false; }

    if (socket_->IsError()) { ok = false; }
  }
  unlock();

  if (!ok) { socket_->close(); }

  return ok;
}

/*
 * Request to take over the connection (socket) from another thread.
 */
bool Connection::take()
{
  bool result = false;
  lock();
  if (!in_use_) {
    in_use_ = true;
    result = true;
  }
  unlock();

  return result;
}

/*
 * Connection Pool
 */
ConnectionPool::ConnectionPool()
{
  connections_ = new alist(10, false);
  /*
   * Initialize mutex and condition variable objects.
   */
  pthread_mutex_init(&add_mutex_, nullptr);
  pthread_cond_init(&add_cond_var_, nullptr);
}

ConnectionPool::~ConnectionPool()
{
  delete (connections_);
  pthread_mutex_destroy(&add_mutex_);
  pthread_cond_destroy(&add_cond_var_);
}

void ConnectionPool::cleanup()
{
  Connection* connection = nullptr;
  int i = 0;
  for (i = connections_->size() - 1; i >= 0; i--) {
    connection = (Connection*)connections_->get(i);
    Dmsg2(800, "checking connection %s (%d)\n", connection->name(), i);
    if (!connection->check()) {
      Dmsg2(120, "connection %s (%d) is terminated => removed\n",
            connection->name(), i);
      connections_->remove(i);
      delete (connection);
    }
  }
}

alist* ConnectionPool::get_as_alist()
{
  cleanup();
  return connections_;
}

bool ConnectionPool::add(Connection* connection)
{
  cleanup();
  Dmsg1(120, "add connection: %s\n", connection->name());
  P(add_mutex_);
  connections_->append(connection);
  pthread_cond_broadcast(&add_cond_var_);
  V(add_mutex_);
  return true;
}

Connection* ConnectionPool::add_connection(const char* name,
                                           int fd_protocol_version,
                                           SusanSocket* socket,
                                           bool authenticated)
{
  Connection* connection =
      new Connection(name, fd_protocol_version, socket, authenticated);
  if (!add(connection)) {
    delete (connection);
    return nullptr;
  }
  return connection;
}

Connection* ConnectionPool::get_connection(const char* name)
{
  Connection* connection = nullptr;
  if (!name) { return nullptr; }
  foreach_alist (connection, connections_) {
    if (connection->check() && connection->authenticated() &&
        connection->bsock() && (!connection->in_use()) &&
        bstrcmp(name, connection->name())) {
      Dmsg1(120, "found connection from client %s\n", connection->name());
      return connection;
    }
  }
  return nullptr;
}

Connection* ConnectionPool::get_connection(const char* name, timespec& timeout)
{
  Connection* connection = nullptr;
  int errstat = 0;

  if (!name) { return nullptr; }

  while ((!connection) && (errstat == 0)) {
    connection = get_connection(name);
    if (!connection) {
      Dmsg0(120, "waiting for new connections.\n");
      errstat = WaitForNewConnection(timeout);
      if (errstat == ETIMEDOUT) {
        Dmsg0(120, "timeout while waiting for new connections.\n");
      }
    }
  }

  return connection;
}

int ConnectionPool::WaitForNewConnection(timespec& timeout)
{
  int errstat;

  P(add_mutex_);
  errstat = pthread_cond_timedwait(&add_cond_var_, &add_mutex_, &timeout);
  V(add_mutex_);
  if (errstat == 0) {
    Dmsg0(120, "new connection available.\n");
  } else if (errstat == ETIMEDOUT) {
    Dmsg0(120, "timeout.\n");
  } else {
    Emsg1(M_ERROR, 0, "error: %d\n", errstat);
  }
  return errstat;
}

bool ConnectionPool::remove(Connection* connection)
{
  bool removed = false;
  for (int i = connections_->size() - 1; i >= 0; i--) {
    if (connections_->get(i) == connection) {
      connections_->remove(i);
      removed = true;
      Dmsg0(120, "removed connection.\n");
      break;
    }
  }
  return removed;
}

Connection* ConnectionPool::remove(const char* name, int timeout_in_seconds)
{
  bool done = false;
  Connection* result = nullptr;
  Connection* connection = nullptr;
  struct timespec timeout;

  ConvertTimeoutToTimespec(timeout, timeout_in_seconds);

  Dmsg2(120, "waiting for connection from client %s. Timeout: %ds.\n", name,
        timeout_in_seconds);

  while (!done) {
    connection = get_connection(name, timeout);
    if (!connection) {
      /*
       * nullptr is returned only on timeout (or other internal errors).
       */
      return nullptr;
    }
    if (connection->take()) {
      result = connection;
      remove(connection);
      done = true;
    } else {
      /*
       * As we can not take it, we assume it is already taken by another thread.
       * In any case, we remove it, to prevent to get stuck in this while loop.
       */
      remove(connection);
    }
  }
  return result;
}
